package com.leilei.join.window;

import com.leilei.join.common.*;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/27 16:52
 * @desc fink 流相连之WindowJoin  (基于窗口的多流相连) 注意，测试发现无界流之间才能join
 * 滑动窗口 设置了已系统时间为时间时间 （滑动窗口必须设置waterMaker）
 */
public class FlinkWindowJoin_WaterMaker {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        // 获取车辆数据
        SingleOutputStreamOperator<Vehicle> vehicleDs = env.addSource(new VehicleSource())
                .assignTimestampsAndWatermarks(new VehicleWatermark());

        // 获取车辆明细数
        SingleOutputStreamOperator<VehicleDetail> vehicleDetailDs = env.addSource(new VehicleDetailSource())
                .assignTimestampsAndWatermarks(new VehicleDetailWatermark());

        // 车辆数据与 车辆明细数据关联滚动五秒统计一次五秒内 关联情况
        DataStream<VehicleInfo> vehicleInfoDs = vehicleDs.join(vehicleDetailDs)
                .where(Vehicle::getId)
                .equalTo(VehicleDetail::getVehicleId)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .apply(new VehicleJoinFunction());

        //再将 vehicleInfo 与车辆类型数据关联
        SingleOutputStreamOperator<VehicleInfo> infoWaterMaker = vehicleInfoDs.
                assignTimestampsAndWatermarks(new VehicleInfoWatermark());

        SingleOutputStreamOperator<Tuple2<Integer, String>> vehicleTypeDs = env.addSource(new VehicleTypeSource())
                .assignTimestampsAndWatermarks(new VehicleTypeWatermark());

        DataStream<Tuple4<Integer, String, Integer, String>> finalSource = infoWaterMaker
                .join(vehicleTypeDs)
                .where(VehicleInfo::getType).equalTo(t -> t.f0)
                .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(7)))
                .apply(new AggregationVehicleJoinFunction());
        finalSource.print("最终车辆数据");
        env.execute("window-join");

    }


}
